package org.lzfto.flink.demo.table.connector.sink;

import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.RowKind;
import org.lzfto.flink.demo.common.JdbcOptionRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public class CustomJdbcOutputFormat implements OutputFormat<RowData> {

    private static Logger LOG = LoggerFactory.getLogger(CustomJdbcOutputFormat.class);
    private static final long serialVersionUID = 1L;

    private final CustomJdbcOutputOperation jdbcOperation;

    private JdbcOptionRecord jdbcOption;

    public CustomJdbcOutputFormat(JdbcOptionRecord  jdbcOption) {
        this.jdbcOption=jdbcOption;
        this.jdbcOperation = new CustomJdbcOutputOperation(jdbcOption);
    }

    @Override    public void configure(Configuration parameters) {
        // 配置初始化
    	//String jsonStr=  JSON.toJSONString(parameters, Feature.PrettyFormat);
    	//System.out.println("配置参数:"+jsonStr);
    }

    @Override
    public void open(int taskNumber, int numTasks) throws IOException {
     
    }

    @Override
    public void writeRecord(RowData record) throws IOException {
        try {
            if (record.getRowKind()== RowKind.INSERT){
                jdbcOperation.executeInsert(record, jdbcOption.dataType());
            }
            else if (record.getRowKind()==RowKind.DELETE){
                jdbcOperation.executeDelete(record,this.jdbcOption.dataType());
            }
            else{
                System.out.println("不支持的操作类型");
            }
        } catch (Exception e) {
            LOG.error("JDBC operation failed", e);
            throw new IOException("JDBC operation failed", e);
        }
    }

    @Override
    public void close() throws IOException {
        // 由JdbcOperationService管理连接生命周期
    }
}
